package com.cloud.easy.im.handler.disruptor;

import com.cloud.easy.im.argeement.Agreement;
import com.cloud.easy.im.event.MessageQoSEventListenerS2C;
import com.cloud.easy.im.event.ServerEventListener;
import com.cloud.easy.im.handler.AbstractServerHandler;
import com.cloud.easy.im.handler.disruptor.event.WebSocketEvent;
import com.cloud.easy.im.handler.strategy.websocket.*;
import com.cloud.easy.im.utils.BasicThreadFactory;
import com.cloud.easy.im.handler.strategy.AgreementStrategy;
import com.google.common.collect.Lists;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.channel.Channel;

import java.util.List;

/**
 * 事件生产者
 *
 * @author wangjian
 * @version 0.1.0
 * @create 2020年10月22日4:58 下午
 */
public class EventProducer {

    /**
     * 环状数据生产桶
     */
    private final RingBuffer<WebSocketEvent> ringBuffer;

    private static final int DEFAULT_BUFFER_SIZE = 2048;

    /**
     * 业务处理器集合
     */
    private List<AgreementStrategy> strategyList = Lists.newArrayList();

    public EventProducer(int bufferSize, ServerEventListener serverEventListener,
                         MessageQoSEventListenerS2C messageQoSEventListenerS2C,
                         AbstractServerHandler abstractServerHandler) {
        strategyList.add(new AgreementWebSocketCCommonPushHandler());
        strategyList.add(new AgreementWebSocketCCommonPullHandler());
        strategyList.add(new AgreementWebSocketCLoginHandler());
        strategyList.add(new AgreementWebSocketCLogoutHandler(abstractServerHandler));
        strategyList.add(new AgreementWebSocketCRecivedHandler(messageQoSEventListenerS2C));

        // 初始化 Disruptor
        Disruptor<WebSocketEvent> disruptor = new Disruptor<WebSocketEvent>(() -> new WebSocketEvent(), bufferSize, new BasicThreadFactory.Builder()
                .namingPattern("EventProducer-daemon-pool-%d").daemon(true).build(),
                ProducerType.SINGLE, new BlockingWaitStrategy());

        disruptor.handleEventsWith(new EventHandler<WebSocketEvent>() {
            @Override
            public void onEvent(WebSocketEvent event, long sequence, boolean endOfBatch) throws Exception {
                for (AgreementStrategy strategy : strategyList) {
                    if (strategy.algorithmInterface(event.getAgreement().getType())) {
                        strategy.handler(event.getSession(), event.getAgreement(), serverEventListener);
                        return;
                    }
                }
            }
        });
        disruptor.setDefaultExceptionHandler(new ExceptionHandler<WebSocketEvent>() {
            @Override
            public void handleEventException(Throwable ex, long sequence, WebSocketEvent event) {
                try {

                } finally {

                }
            }

            @Override
            public void handleOnStartException(Throwable ex) {

            }

            @Override
            public void handleOnShutdownException(Throwable ex) {

            }
        });

        disruptor.start();
        ringBuffer = disruptor.getRingBuffer();
    }

    public void onHandler(Channel session, Agreement agreement) {
        long sequence = ringBuffer.next();
        try {
            WebSocketEvent webSocketEvent = ringBuffer.get(sequence);
            webSocketEvent.setAgreement(agreement);
            webSocketEvent.setSession(session);
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}
